package study.stream

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * @author zh
 * @date 2021/6/17 15:29
 */
object CreateStreamContext {
  def main(args: Array[String]): Unit = {
    val checkPointDir = "./checkpoint3"

    // 驱动器程序容错
    val context = StreamingContext.getOrCreate(checkPointDir, () => {
      val sparkConf = new SparkConf().setMaster("local[*]").setAppName("sparkStreaming")
      val ssc = new StreamingContext(sparkConf, Seconds(1))
      ssc.checkpoint(checkPointDir)
      // 业务逻辑要放在该方法内部，否则无法生效
//      val stream = ssc.textFileStream("./data/stream.txt")
      val stream = ssc.receiverStream(new MyReceiver)
      val map = stream.map((_, 1))


      val window = map.reduceByKeyAndWindow(
        { (x, y) => x + y }, // 加上新进入窗口的批次中的元素
        { (x, y) => x - y }, //移除离开窗口的老批次中的元素
        Seconds(10), //窗口时长
        Seconds(1) //滑动步长
      )

      window.print()

      ssc
    })




    // 启动采集器
    context.start()

    // 等待采集器停止
    context.awaitTermination()

  }

}
